package cn.sheep.dmp.report

import cn.sheep.dmp.tools.JPools
import cn.sheep.dmp.utils.{CaluAdKpi, Tools}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Sheep.Old @ 64341393
  * Created 2018/3/29
  */
object ReportMediaRedis {

    def main(args: Array[String]): Unit = {


        val sparkConf = new SparkConf().setAppName("媒体纬度指标分布")
          .setMaster("local[*]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(sparkConf)
        val sqlc = new SQLContext(sc)

        // 读parquet文件
        val dataFrame = sqlc.read.parquet(Tools.load.getString("parquet.path"))


        // 注册一个的函数
        sqlc.udf.register("isNotEmpty", (appId: String, appName: String) => {
            if (StringUtils.isNotEmpty(appId) || StringUtils.isNotEmpty(appName)) true else false
        })

        // DSL
        dataFrame.filter("isNotEmpty(appid, appname)").mapPartitions(iter => {

            val jedis = JPools.getJedis

            val result = iter.map(row => {
                val appId = row.getAs[String]("appid")
                var appName = row.getAs[String]("appname")

                if (StringUtils.isEmpty(appName)) { // 如果appName为空则去字典中找

                    appName = jedis.hget("appdict", appId)
                    if (StringUtils.isEmpty(appName)) appName = appId
                }

                val rmode = row.getAs[Int]("requestmode")
                val pnode = row.getAs[Int]("processnode")
                val efft = row.getAs[Int]("iseffective")
                val bill = row.getAs[Int]("isbilling")
                val isbid = row.getAs[Int]("isbid")
                val iswin = row.getAs[Int]("iswin")
                val adoid = row.getAs[Int]("adorderid")

                val cb = row.getAs[Double]("winprice")
                val xf = row.getAs[Double]("adpayment")


                val adReq: List[Double] = CaluAdKpi.caluAdReq(rmode, pnode)
                val adRtb: List[Double] = CaluAdKpi.caluAdRbt(efft, bill, isbid, iswin, adoid)
                val adShowClick: List[Double] = CaluAdKpi.caluAdShowClick(rmode, efft)
                val adCost: List[Double] = CaluAdKpi.caluAdCost(efft, bill, iswin, cb, xf)

                (appName, adReq ++ adRtb ++ adShowClick ++ adCost)
            })
            jedis.close()
            result

        }).reduceByKey {
            case (list1, list2) => list1.zip(list2).map(t => t._1 + t._2)
        }.sortBy[Double](tp => tp._2.head, false)
          .saveAsTextFile("F:\\dmp\\RptApp2")


        sc.stop()
    }

}
